package com.atguigu.bigdata.spark.core.rdd.persist

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  * create by undeRdoG on  2021-06-14  14:07
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object Spark03_RDD_Persist {


  /**
    * ***************************************************************************
    * ***************************************************************************
    * ***************************************************************************
    * ***************************************************************************
    * (Spark,1)
    * (Hello,2)
    * (Scala,1)
    * (Spark,CompactBuffer(1))
    * (Hello,CompactBuffer(1, 1))
    * (Scala,CompactBuffer(1))
    *
    * 对 RDD 进行cache  该RDD计算时的数据会被持久化
    *
    **/
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(sparkConf)

    val list = List("Hello Scala", "Hello Spark")

    val rdd: RDD[String] = sc.makeRDD(list)

    val flatRDD = rdd.flatMap(_.split(" "))

    val mapRDD = flatRDD.map(x => {
      println("****************************************************************************")
      (x, 1)
    })

    /**
      * 持久化 RDD 计算时的数据   默认持久化到内存
      * 可以选择 其他存级别
      **/
    mapRDD.cache()

    /**
      * 持久化操作必须在 行动算子执行时才能完成
      *
      * RDD 持久化操作，不一定为了重用。在数据比较重要的场合，也可以进行持久化
      **/
    mapRDD.persist(StorageLevel.DISK_ONLY)


    val res: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)

    res.collect().foreach(println)


    val res2: RDD[(String, Iterable[Int])] = mapRDD.groupByKey()

    res2.collect().foreach(println)

  }
}
